package com.example.rxbyhand;

import com.example.rxbyhand.scheduler.Scheduler;

/**
 * Author by lp,on 2021/8/17/017,11:44.
 */
public class ObservableSubscribeOn<T> extends AbstractObservanleWithUpStream<T, T> {
    final Scheduler scheduler;

    public ObservableSubscribeOn(ObservableSource<T> source, Scheduler scheduler) {
        super(source);
        this.scheduler = scheduler;
    }


    @Override
    protected void subscribeActual(Observer observer) {
        Scheduler.Worker worker=scheduler.creteWorker();
        worker.schedule(new SubscribeTask(new SubscribeOnObserver<T>(observer)));

    }

    class SubscribeOnObserver<T> implements Observer<T> {

        final Observer<T> downStream;

        SubscribeOnObserver(Observer<T> downStream) {
            this.downStream = downStream;

        }

        @Override
        public void onSubscribe() {
            downStream.onSubscribe();

        }

        @Override
        public void onNext(T t) {
            downStream.onNext(t);
        }

        @Override
        public void onComplete() {
            downStream.onComplete();
        }

        @Override
        public void onError(Throwable throwable) {
            downStream.onError(throwable);
        }
    }

    final class SubscribeTask implements Runnable {

        final SubscribeOnObserver partent;

        SubscribeTask(SubscribeOnObserver<T> partent) {
            this.partent = partent;
        }

        @Override
        public void run() {
            source.subscribe(partent);
        }
    }
}
